package com.exsun.gateway.jt809.kafka;

import com.exsun.gateway.jt809.common.DevInfoService;
import com.exsun.gateway.jt809.data.PushData;
import com.exsun.gateway.jt809.handler.cache.MediaHandle;
import com.exsun.gateway.jt809.kafka.gps.GpsConvert;
import com.exsun.gateway.jt809.config.Jt809Config;
import com.exsun.gateway.jt809.handler.SendMgrHandler;
import com.google.gson.reflect.TypeToken;
import exsun.jt808.data.config.jt808.Jt808ReceiveMsgTopic;
import exsun.jt808.data.config.jt809.Jt809MsgIdConfig;
import exsun.jt808.data.jt809_data.monitor.main.UpCtrlMsgTakePhotoAck;
import exsun.jt808.data.jt809_data.monitor.slave.DownCtrlMsgTakePhotoReq;
import exsun.jt808.data.kafka_data.KafkaJT808Data;
import exsun.jt808.data.upload_data.jt808.MediaFullData;
import exsun.jt808.data.util.FtpUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;
import org.apache.commons.io.FilenameUtils;
import java.lang.reflect.Type;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

@Service
public class JT809KafkaMediaHandler extends JT809KafkaHandleBase<MediaFullData>{
    private Logger logger = LoggerFactory.getLogger(JT809KafkaMediaHandler.class);

    @Autowired
    private SendMgrHandler sendMgrHandler;

    @Autowired
    private Jt809Config jt809Config;

    @Autowired
    private DevInfoService devInfoService;

    //订阅媒体数据
    @KafkaListener(topics = Jt808ReceiveMsgTopic.mediaFullData, containerFactory = "mediaDataFactory")
    public void batchMediaMsgConsumerListen(List<ConsumerRecord<?, ?>> records, Acknowledgment ack) {
        try{
            Date dt = new Date();
            logger.info("接收多媒体数据..");
            //多媒体数据上传
            Type type = new TypeToken<KafkaJT808Data<MediaFullData>>(){}.getType();
            List<PushData> pushs =  super.done(records, type, (obj)->{
                UpCtrlMsgTakePhotoAck upCtrlMsgTakePhotoAck = getUpCtrlMsgTakePhotoAck(obj);
                if(upCtrlMsgTakePhotoAck == null){
                    logger.info("该图片未匹配上指令，不上传");
                    return null;
                }

                byte[] body = sendMgrHandler.jt809Encode.upCtrlMsgTakePhotoAck(upCtrlMsgTakePhotoAck);
                PushData pushData = new PushData(Jt809MsgIdConfig.UP_CTRL_MSG, body);

                List<PushData> temp = new ArrayList<>();
                temp.add(pushData);
                return temp;

            });
            //发送错误，则不提交偏移量
            if(pushs == null)return;
            if(pushs.size() != 0){
                logger.info("上报多媒体：{}条", pushs.size());
            }
            long span = new Date().getTime() - dt.getTime();
            logger.info("处理[{}]数据，花费时间{}ms",type.getTypeName(), span);
            ack.acknowledge();
        }catch(Exception e){
            logger.info(e.getMessage());
        }
    }

    public UpCtrlMsgTakePhotoAck getUpCtrlMsgTakePhotoAck(MediaFullData data) {
        MediaFullData mediaFullData = data;
        //获取缓存的指令数据， 推送上级平台
        DownCtrlMsgTakePhotoReq req = MediaHandle.inst.get(mediaFullData.dvo,mediaFullData.getChannelId());
        if(req == null){
            logger.info("上报多媒体失败：DownCtrlMsgTakePhotoReq == null");
            return null;
        }
        UpCtrlMsgTakePhotoAck upCtrlMsgTakePhotoAck = uploadMediaInfo(mediaFullData, req);
        return upCtrlMsgTakePhotoAck;
    }

    /**
     * 上传多媒体
     *
     */
    private UpCtrlMsgTakePhotoAck uploadMediaInfo(MediaFullData media, DownCtrlMsgTakePhotoReq req) {
        UpCtrlMsgTakePhotoAck upCtrlMsgTakePhotoAck = new UpCtrlMsgTakePhotoAck();
        upCtrlMsgTakePhotoAck.setBasicVehicleInfo(req.getBasicVehicleInfo());
        upCtrlMsgTakePhotoAck.setPhotoRspFlag(1);
        upCtrlMsgTakePhotoAck.setGnssData(GpsConvert.gnssDataHandle(media.getGpsData()));
        upCtrlMsgTakePhotoAck.setLensId(req.getLensId());
        upCtrlMsgTakePhotoAck.setSizeType(req.getSize());
        upCtrlMsgTakePhotoAck.setType(media.getMediaType());
//        upCtrlMsgTakePhotoAck.setPhoto(media.getMediaBytes());
        //因为kafka推送大量数据会报错，故改为从文件路径读取图片数据
        String path = media.getMediaPath();

        String fileName = FilenameUtils.getName(path);
        String filePath = FilenameUtils.getPath(path);
        logger.info("读取ftp图片数据....{}",path);
        byte[] buffer =  getByFTP(fileName, filePath);
        logger.info("读取ftp图片数据成功 size:{} {}",buffer.length, path);
        upCtrlMsgTakePhotoAck.setPhoto(buffer);
        return upCtrlMsgTakePhotoAck;
    }

    public byte[] getByFTP(String fileName, String filePath){
        try {
            String remotePath = jt809Config.getJt809FtpBasePath()+"/"+ filePath;
            byte[] buffer = FtpUtils.downloadFile(jt809Config.getJt809FtpServer(),
                    jt809Config.getJt809FtpPort(), jt809Config.getJt809FtpUserName(),
                    jt809Config.getJt809FtpPassword(),remotePath,fileName,jt809Config.getJt809FtpTimeOut());
            return buffer;
        }catch (Exception e){
            logger.info("读取ftp失败--" + "ftp地址" + jt809Config.getJt809FtpBasePath()+"/"+ filePath + "--错误信息--" + e.getMessage());
            return null;
        }
    }
}
